package com.shujia.mr.kqzl;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.ArrayList;

/**
 * <"1001-0809",60>
 */

public class kqzlMaper extends Mapper<LongWritable, Text, Text, LongWritable> {
    private ArrayList<String> idlist  = new ArrayList<>();
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException, InterruptedException {
        //map是按照行进行读取的
        //我们数据的第一行是标题，涵盖了监测点的编号
        String line = value.toString();
        if (line.startsWith("date")) {//切的是标题
            //如果一行以data开头的说明这行是标题
            String[] infos = line.split(",");
            //切出来的infos数组是存放标题的
            for (int i = 3; i < infos.length; i++) {
                idlist.add(infos[i].substring(0, infos[i].length() - 1));

            }
        }else {
            String[] infos = line.split(",");
            //判断这一行是不是采集PM2.5
            if ("PM2.5".equals(infos[2])){
                //空数据的处理
                for (int i = 3,j=0; i < infos.length && j<idlist.size(); i++,j++) {
                    if (infos[i] == null || "".equals(infos[i])) {
                        infos[i] = "0";
                    }//日期-区域
                    context.write(new Text(infos[0]+"-"+idlist.get(j)),new LongWritable(Long.parseLong(infos[i])));//拼接字符串，输出格式
                }

            }
        }


    }
}
